package com.ferry.taskflow.taskFlow.nontx;


import com.ferry.taskflow.taskFlow.base.IdAble;
import com.ferry.taskflow.taskFlow.base.enumv.StageResult;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.*;

/**
 * Created by daiyong
 * 一个任务流程
 * fLow
 * ---- stage1， stage2
 * ------------- task1 task2
 */
@Slf4j
public class NonTxFlow extends IdAble {

	private List<NonTxStage> nonTxStages = Lists.newArrayList();

	private ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) Executors.newCachedThreadPool();

	private NonTxStage lastNonTxStage = null;

	private NonTxFlow(){
		super();
	}

	public boolean start(){
		Assert.notEmpty(nonTxStages, "flow阶段不能为空");

        //顺序执行stage
		for (int i = 0; i < nonTxStages().size(); i++) {
			log.info("{}:flow 执行第 {} 个阶段", id(), i + 1);

			NonTxStage stage = nonTxStages().get(i);
			NonTxStage preStage = stage.preStage();

			if (null != preStage) {
				if (!stage.ignorePreStageException() && preStage.stageResult() == StageResult.EXCEPTION) {
                     log.error("{}:flow 执行阶段异常,flow停止, {}", id(), new JsonEncoder().encode(stage));
					 return false;
				}

				if (!stage.ignorePreStageTimeout() && preStage.stageResult() == StageResult.TIMEOUT) {
					log.error("{}:flow 执行阶段超时,flow停止, {}", id(), new JsonEncoder().encode(stage));
					return false;
				}
			}

			//默认并行执行 串行没有意义

			List<NonTxTaskExecutor> executors = stage.executors();

			CompletableFuture[] futures = new CompletableFuture[executors.size()];
			for (int j = 0; j < executors.size(); j++) {
				NonTxTaskExecutor nonTxTaskExecutor = executors.get(j);
				futures[j] = CompletableFuture.runAsync(() -> nonTxTaskExecutor.execute(), threadPoolExecutor);
			}

			//需要全部执行完毕，才能进行下一阶段
			try {
				if (stage.isAll()) {
					if (stage.isTimeout()) {
						CompletableFuture.allOf(futures).get(stage.timeout(), TimeUnit.SECONDS);
					} else {
						CompletableFuture.allOf(futures).get();
					}
				}
				//任意任务执行完毕，就可以进行下一阶段
				else {
					if (stage.isTimeout()) {
						CompletableFuture.anyOf(futures).get(stage.timeout(), TimeUnit.SECONDS);
					} else {
						CompletableFuture.anyOf(futures).get();
					}
				}
			} catch (TimeoutException e) {
				log.error("{}:flow {}:stage 执行超时", id(), stage.id(), e);
				stage.stageResult(StageResult.EXCEPTION);
			} catch (Exception e) {
				log.error("{}:flow {}:stage 执行异常", id(), stage.id(), e);
				stage.stageResult(StageResult.TIMEOUT);
			}

		}

		return true;
	}

	public void shutdown() {
		this.threadPoolExecutor.shutdown();
	}

	/**
	 * 并行执行
	 * @param nonTxStage
	 */

	public NonTxFlow addNonTxStage(NonTxStage nonTxStage) {
		nonTxStage.preStage(lastNonTxStage);
		lastNonTxStage = nonTxStage;
		nonTxStages.add(nonTxStage);
		return this;
	}

	public List<NonTxStage> nonTxStages() {
		return this.nonTxStages;
	}

	public NonTxFlow threadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
		this.threadPoolExecutor = threadPoolExecutor;
		return this;
	}

	public ThreadPoolExecutor threadPoolExecutor() {
		return this.threadPoolExecutor;
	}

	public static NonTxFlow create() {
		return new NonTxFlow();
	}


}
